#include "app_pool.h"

typedef struct
{
    void (*fun)(void *); //任务
    void *data;//任务的参数
} Pool_Task;

static const char *MQ_NAME = "/app_pool_mq";
static pthread_t *threads; // 线程标识符
static int thread_num;     // 线程池数量
static mqd_t mqd;          // 消息队列描述符

// 每个线程都是单独的栈空间
static void *thread_func(void *arg)
{
    Pool_Task pool_Task;
    while (1)
    {
        // 接收消息，接收不到会阻塞
        ssize_t len = mq_receive(mqd, (char *)&pool_Task, sizeof(Pool_Task), NULL);
        if (len != -1)
        {
            pool_Task.fun(pool_Task.data);
        }
    }
}
Com_State app_pool_init(u_int8_t size)
{
    // 1. 创建消息队列 放在第一步，消息队列没创建成功，线程池也无意义
    struct mq_attr attr = {
        .mq_flags = 0,
        .mq_maxmsg = 10,	//注意长度不能超过10，如果超过，可通过 /proc/sys/fs/mqueue/ 目录下的msg_max配置
        .mq_msgsize = sizeof(Pool_Task)};
    mqd = mq_open(MQ_NAME, O_RDWR | O_CREAT, 0666, &attr);
    if (mqd == -1)
    {
        log_error("消息队列打开/创建失败");
        return COM_ERROR;
    }
    // 2. 创建线程池
    thread_num = size;
    // 开辟一块内存，存放线程的标识符，因为这个变量首先要放在全局，别的地方用，其次没调用init不知道长度是多少
    threads = malloc(sizeof(pthread_t) * size);
    if (threads == NULL)
    {
        log_error("内存分配失败");
        mq_close(mqd);
        mq_unlink(MQ_NAME);
        return COM_ERROR;
    }
    for (u_int8_t i = 0; i < size; i++)
    {
        int res = pthread_create(&threads[i], NULL, thread_func, NULL);
        if (res == -1)
        {
            log_error("线程创建失败");
            // 如果想要详细点还要看创建第几个线程失败，还要处理，是重新创建还是deinit，
            // 但是deinit不能是thread_num，应该是已经创建的线程数量
            return COM_ERROR;
        }
    }
    return COM_OK;
}

void app_pool_deinit(void)
{
    for (int i = 0; i < thread_num; i++)
    {
        pthread_cancel(threads[i]);     // pthread_cancel() 是非阻塞的，只是向线程发送取消信号
        pthread_join(threads[i], NULL); // 线程收到取消信号后，需要等待结束
        // 如果不加join可能出现，执行free(threads)，但是清理工作访问了已经释放的thread
        // 执行了close，但访问了已经关闭的消息队列
    }
    free(threads);
    mq_close(mqd);
    mq_unlink(MQ_NAME);
    log_info("线程池销毁");
}

Com_State app_pool_register_task(void (*func)(void *), void *argv)
{
    if (func == NULL)
    {
        log_error("必须传递函数指针");
        return COM_ERROR;
    }
    Pool_Task task = {
        .fun = func,
        .data = argv};
    int res = mq_send(mqd,(char *) &task, sizeof(Pool_Task), 0);
    return res == 0 ? COM_OK : COM_ERROR;
}
